package com.shujia.flink.core

import org.apache.flink.api.common.functions.JoinFunction
import org.apache.flink.streaming.api.functions.co.CoMapFunction
import org.apache.flink.streaming.api.scala._
import org.apache.flink.streaming.api.windowing.assigners.TumblingProcessingTimeWindows
import org.apache.flink.streaming.api.windowing.time.Time

object Demo11Join {
  def main(args: Array[String]): Unit = {
    /**
      * 动态拦截黑名单
      *
      */

    val env = StreamExecutionEnvironment.getExecutionEnvironment

    /**
      * id,date
      * 1,20200510
      * 2,20200511
      * 3,20200512
      * 4,20200513
      *
      */
    val ds = env.socketTextStream("node1", 6666)
      .map(line => {
        val split = line.split(",")
        (split(0), split(1))
      })

    /**
      * id,action
      * 1,0
      * 2,2
      * 3,1
      * 4,2
      *
      */
    val ds1 = env.socketTextStream("node1", 7777)
      .map(line => {
        val split = line.split(",")
        (split(0), split(1))
      })


    val joinDS = ds
      .join(ds1)
      .where(_._1) //指定第一个流关联条件
      .equalTo(_._1) //指定第二个流关联条件
      .window(TumblingProcessingTimeWindows.of(Time.seconds(10)))
      .apply(new JoinFunction[(String, String), (String, String), (String, String, String)] {
        override def join(first: (String, String), second: (String, String)): (String, String, String) = {
          (first._1, first._2, second._2)
        }
      })

    joinDS.print()


    env.execute()
  }
}


